package net.wlm.realtime.new_job;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import org.apache.flink.api.common.functions.RichMapFunction;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.datastream.KeyedStream;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.functions.ProcessFunction;
import org.apache.flink.table.api.EnvironmentSettings;
import org.apache.flink.table.api.TableEnvironment;
import org.apache.flink.util.Collector;
import org.apache.flink.util.OutputTag;

public class RealtimeVehicleMonitorSql {
    public static void main(String[] args) {
        // 1. 表执行环境
        TableEnvironment tabEnv = getTableEnv();
        // 2. 输入表-input:映射到kafka消息队列
        createInputTable(tabEnv);
        // 3.数据处理
       // handle(tabEnv);
        // 3. 输出表-output:映射到clickhouse
       // createOutputTable(tabEnv);
        // 4. 保存数据
       // saveToClickHouse(tabEnv);
    }

    private static void saveToClickHouse(TableEnvironment tabEnv) {

    }
    private static void createOutputTable(TableEnvironment tabEnv) {
    }
    private static void handle(TableEnvironment tabEnv) {

    }
    private static void createInputTable(TableEnvironment tabEnv) {
        tabEnv.executeSql("CREATE TABLE vehicle_log_kafka_sink (\n" +
                "  vin               varchar(32) comment '车辆唯一标识',\n" +
                "  `timestamp` timestamp(3) comment '数据时间戳',\n" +
                "  longitude         float comment '经度',\n" +
                "  latitude          float comment '维度',\n" +
                "  speed             float comment '车速(km/h)',\n" +
                "  mileage           float comment '总里程(km)',\n" +
                "  batteryLevel      float comment '电池电量(%)',\n" +
                "  batteryTemp       float comment '电池温度(℃)',\n" +
                "  motorTemp         float comment '电机温度(℃)',\n" +
                "  chargingStatus    varchar(8) comment '充电状态：charging(充电)、discharging(放电)',\n" +
                "  energyConsumption float comment '瞬时能耗'\n" +
                ")WITH (\n" +
                "     'connector' = 'kafka',\n" +
                "     'topic' = 'vehicle-data',\n" +
                "     'properties.bootstrap.servers' = 'node101:9092,node102:9092,node103:9092',\n" +
                "     'properties.group.id' = 'gid-dwd-order-detail',\n" +
                "     'scan.startup.mode' = 'earliest-offset',\n" +
                "     'format' = 'json',\n" +
                "     'json.fail-on-missing-field' = 'false',\n" +
                "     'json.ignore-parse-errors' = 'true'\n" +
                "     )");
        tabEnv.sqlQuery("select * from vehicle_log_kafka_sink").execute().print();
    }


    private static TableEnvironment getTableEnv() {
        // a-环境属性设置
        EnvironmentSettings settings = EnvironmentSettings.newInstance()
                .inStreamingMode()
                .useBlinkPlanner()
                .build();
        TableEnvironment tabEnv = TableEnvironment.create(settings);
        // b-属性配置设置
        Configuration configuration = tabEnv.getConfig().getConfiguration();
        configuration.setString("table.local-time-zone", "Asia/Shanghai");
        configuration.setString("table.exec.resource.default-parallelism", "1");
        configuration.setString("table.exec.state.ttl", "5 s");
        // c-返回对象
        return tabEnv;
    }
}